//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//

package org.apache.rocketmq.spring.support;

import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.utils.MessageUtil;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.rocketmq.spring.core.RocketMQReplyListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.MethodParameter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.converter.SmartMessageConverter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.MimeTypeUtils;

public class DefaultRocketMQListenerContainer implements InitializingBean, RocketMQListenerContainer, SmartLifecycle, ApplicationContextAware {
    private static final Logger log = LoggerFactory.getLogger(DefaultRocketMQListenerContainer.class);
    private ApplicationContext applicationContext;
    private String name;
    private long suspendCurrentQueueTimeMillis = 1000L;
    private int delayLevelWhenNextConsume = 0;
    private String nameServer;
    private AccessChannel accessChannel;
    private String consumerGroup;
    private String topic;
    private int consumeThreadMax;
    private int consumeThreadNumber;
    private String charset;
    private MessageConverter messageConverter;
    private RocketMQListener rocketMQListener;
    private RocketMQReplyListener rocketMQReplyListener;
    private RocketMQMessageListener rocketMQMessageListener;
    private DefaultMQPushConsumer consumer;
    private Type messageType;
    private MethodParameter methodParameter;
    private boolean running;
    private ConsumeMode consumeMode;
    private SelectorType selectorType;
    private String selectorExpression;
    private MessageModel messageModel;
    private long consumeTimeout;
    private int maxReconsumeTimes;
    private int replyTimeout;
    private String tlsEnable;
    private String namespace;
    private long awaitTerminationMillisWhenShutdown;
    private String instanceName;

    public DefaultRocketMQListenerContainer() {
        this.accessChannel = AccessChannel.LOCAL;
        this.consumeThreadMax = 64;
        this.consumeThreadNumber = 20;
        this.charset = "UTF-8";
    }

    public long getSuspendCurrentQueueTimeMillis() {
        return this.suspendCurrentQueueTimeMillis;
    }

    public void setSuspendCurrentQueueTimeMillis(long suspendCurrentQueueTimeMillis) {
        this.suspendCurrentQueueTimeMillis = suspendCurrentQueueTimeMillis;
    }

    public int getDelayLevelWhenNextConsume() {
        return this.delayLevelWhenNextConsume;
    }

    public void setDelayLevelWhenNextConsume(int delayLevelWhenNextConsume) {
        this.delayLevelWhenNextConsume = delayLevelWhenNextConsume;
    }

    public String getNameServer() {
        return this.nameServer;
    }

    public void setNameServer(String nameServer) {
        this.nameServer = nameServer;
    }

    public AccessChannel getAccessChannel() {
        return this.accessChannel;
    }

    public void setAccessChannel(AccessChannel accessChannel) {
        this.accessChannel = accessChannel;
    }

    public String getConsumerGroup() {
        return this.consumerGroup;
    }

    public void setConsumerGroup(String consumerGroup) {
        this.consumerGroup = consumerGroup;
    }

    public String getTopic() {
        return this.topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public int getConsumeThreadMax() {
        return this.consumeThreadMax;
    }

    public int getConsumeThreadNumber() {
        return this.consumeThreadNumber;
    }

    public String getCharset() {
        return this.charset;
    }

    public void setCharset(String charset) {
        this.charset = charset;
    }

    public MessageConverter getMessageConverter() {
        return this.messageConverter;
    }

    public DefaultRocketMQListenerContainer setMessageConverter(MessageConverter messageConverter) {
        this.messageConverter = messageConverter;
        return this;
    }

    public RocketMQListener getRocketMQListener() {
        return this.rocketMQListener;
    }

    public void setRocketMQListener(RocketMQListener rocketMQListener) {
        this.rocketMQListener = rocketMQListener;
    }

    public RocketMQReplyListener getRocketMQReplyListener() {
        return this.rocketMQReplyListener;
    }

    public void setRocketMQReplyListener(RocketMQReplyListener rocketMQReplyListener) {
        this.rocketMQReplyListener = rocketMQReplyListener;
    }

    public RocketMQMessageListener getRocketMQMessageListener() {
        return this.rocketMQMessageListener;
    }

    public void setRocketMQMessageListener(RocketMQMessageListener anno) {
        this.rocketMQMessageListener = anno;
        this.consumeMode = anno.consumeMode();
        this.consumeThreadMax = anno.consumeThreadMax();
        this.consumeThreadNumber = anno.consumeThreadNumber();
        this.messageModel = anno.messageModel();
        this.selectorType = anno.selectorType();
        this.selectorExpression = anno.selectorExpression();
        this.consumeTimeout = anno.consumeTimeout();
        this.maxReconsumeTimes = anno.maxReconsumeTimes();
        this.replyTimeout = anno.replyTimeout();
        this.tlsEnable = anno.tlsEnable();
        this.namespace = anno.namespace();
        this.delayLevelWhenNextConsume = anno.delayLevelWhenNextConsume();
        this.suspendCurrentQueueTimeMillis = (long)anno.suspendCurrentQueueTimeMillis();
        this.awaitTerminationMillisWhenShutdown = (long)Math.max(0, anno.awaitTerminationMillisWhenShutdown());
        this.instanceName = anno.instanceName();
    }

    public ConsumeMode getConsumeMode() {
        return this.consumeMode;
    }

    public SelectorType getSelectorType() {
        return this.selectorType;
    }

    public void setSelectorExpression(String selectorExpression) {
        this.selectorExpression = selectorExpression;
    }

    public String getSelectorExpression() {
        return this.selectorExpression;
    }

    public MessageModel getMessageModel() {
        return this.messageModel;
    }

    public String getTlsEnable() {
        return this.tlsEnable;
    }

    public void setTlsEnable(String tlsEnable) {
        this.tlsEnable = tlsEnable;
    }

    public String getNamespace() {
        return this.namespace;
    }

    public void setNamespace(String namespace) {
        this.namespace = namespace;
    }

    public DefaultMQPushConsumer getConsumer() {
        return this.consumer;
    }

    public void setConsumer(DefaultMQPushConsumer consumer) {
        this.consumer = consumer;
    }

    public long getAwaitTerminationMillisWhenShutdown() {
        return this.awaitTerminationMillisWhenShutdown;
    }

    public String getInstanceName() {
        return this.instanceName;
    }

    public void setInstanceName(String instanceName) {
        this.instanceName = instanceName;
    }

    public DefaultRocketMQListenerContainer setAwaitTerminationMillisWhenShutdown(long awaitTerminationMillisWhenShutdown) {
        this.awaitTerminationMillisWhenShutdown = awaitTerminationMillisWhenShutdown;
        return this;
    }

    public void destroy() {
        this.setRunning(false);
        if (Objects.nonNull(this.consumer)) {
            this.consumer.shutdown();
        }

        log.info("container destroyed, {}", this.toString());
    }

    public boolean isAutoStartup() {
        return true;
    }

    public void stop(Runnable callback) {
        this.stop();
        callback.run();
    }

    public void start() {
        if (this.isRunning()) {
            throw new IllegalStateException("container already running. " + this.toString());
        } else {
            try {
                this.consumer.start();
            } catch (MQClientException var2) {
                throw new IllegalStateException("Failed to start RocketMQ push consumer", var2);
            }

            this.setRunning(true);
            log.info("running container: {}", this.toString());
        }
    }

    public void stop() {
        if (this.isRunning()) {
            if (Objects.nonNull(this.consumer)) {
                this.consumer.shutdown();
            }

            this.setRunning(false);
        }

    }

    public boolean isRunning() {
        return this.running;
    }

    private void setRunning(boolean running) {
        this.running = running;
    }

    public int getPhase() {
        return Integer.MAX_VALUE;
    }

    public void afterPropertiesSet() throws Exception {
        this.initRocketMQPushConsumer();
        this.messageType = this.getMessageType();
        this.methodParameter = this.getMethodParameter();
        log.debug("RocketMQ messageType: {}", this.messageType);
    }

    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    public String toString() {
        return "DefaultRocketMQListenerContainer{consumerGroup='" + this.consumerGroup + '\'' + ", namespace='" + this.namespace + '\'' + ", nameServer='" + this.nameServer + '\'' + ", topic='" + this.topic + '\'' + ", consumeMode=" + this.consumeMode + ", selectorType=" + this.selectorType + ", selectorExpression='" + this.selectorExpression + '\'' + ", messageModel=" + this.messageModel + '\'' + ", tlsEnable=" + this.tlsEnable + ", instanceName=" + this.instanceName + '}';
    }

    public void setName(String name) {
        this.name = name;
    }

    public void handleMessage(MessageExt messageExt) throws MQClientException, RemotingException, InterruptedException {
        if (this.rocketMQListener != null) {
            this.rocketMQListener.onMessage(this.doConvertMessage(messageExt));
        } else if (this.rocketMQReplyListener != null) {
            Object replyContent = this.rocketMQReplyListener.onMessage(this.doConvertMessage(messageExt));
            Message<?> message = MessageBuilder.withPayload(replyContent).build();
            org.apache.rocketmq.common.message.Message replyMessage = MessageUtil.createReplyMessage(messageExt, this.convertToBytes(message));
            DefaultMQProducer producer = this.consumer.getDefaultMQPushConsumerImpl().getmQClientFactory().getDefaultMQProducer();
            producer.setSendMsgTimeout(this.replyTimeout);
            producer.send(replyMessage, new SendCallback() {
                public void onSuccess(SendResult sendResult) {
                    if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                        DefaultRocketMQListenerContainer.log.error("Consumer replies message failed. SendStatus: {}", sendResult.getSendStatus());
                    } else {
                        DefaultRocketMQListenerContainer.log.debug("Consumer replies message success.");
                    }

                }

                public void onException(Throwable e) {
                    DefaultRocketMQListenerContainer.log.error("Consumer replies message failed. error: {}", e.getLocalizedMessage());
                }
            });
        }

    }

    private byte[] convertToBytes(Message<?> message) {
        Message<?> messageWithSerializedPayload = this.doConvert(message.getPayload(), message.getHeaders());
        Object payloadObj = messageWithSerializedPayload.getPayload();

        try {
            if (null == payloadObj) {
                throw new RuntimeException("the message cannot be empty");
            } else {
                byte[] payloads;
                if (payloadObj instanceof String) {
                    payloads = ((String)payloadObj).getBytes(Charset.forName(this.charset));
                } else if (payloadObj instanceof byte[]) {
                    payloads = (byte[])((byte[])messageWithSerializedPayload.getPayload());
                } else {
                    String jsonObj = (String)this.messageConverter.fromMessage(messageWithSerializedPayload, payloadObj.getClass());
                    if (null == jsonObj) {
                        throw new RuntimeException(String.format("empty after conversion [messageConverter:%s,payloadClass:%s,payloadObj:%s]", this.messageConverter.getClass(), payloadObj.getClass(), payloadObj));
                    }

                    payloads = jsonObj.getBytes(Charset.forName(this.charset));
                }

                return payloads;
            }
        } catch (Exception var6) {
            throw new RuntimeException("convert to bytes failed.", var6);
        }
    }

    private Message<?> doConvert(Object payload, MessageHeaders headers) {
        Message<?> message = this.messageConverter instanceof SmartMessageConverter ? ((SmartMessageConverter)this.messageConverter).toMessage(payload, headers, (Object)null) : this.messageConverter.toMessage(payload, headers);
        if (message == null) {
            String payloadType = payload.getClass().getName();
            Object contentType = headers != null ? headers.get("contentType") : null;
            throw new MessageConversionException("Unable to convert payload with type='" + payloadType + "', contentType='" + contentType + "', converter=[" + this.messageConverter + "]");
        } else {
            MessageBuilder<?> builder = MessageBuilder.fromMessage(message);
            builder.setHeaderIfAbsent("contentType", MimeTypeUtils.TEXT_PLAIN);
            return builder.build();
        }
    }

    private Object doConvertMessage(MessageExt messageExt) {
        if (!Objects.equals(this.messageType, MessageExt.class) && !Objects.equals(this.messageType, org.apache.rocketmq.common.message.Message.class)) {
            String str = new String(messageExt.getBody(), Charset.forName(this.charset));
            if (Objects.equals(this.messageType, String.class)) {
                return str;
            } else {
                try {
                    return this.messageType instanceof Class ? this.getMessageConverter().fromMessage(MessageBuilder.withPayload(str).build(), (Class)this.messageType) : ((SmartMessageConverter)this.getMessageConverter()).fromMessage(MessageBuilder.withPayload(str).build(), (Class)((ParameterizedType)this.messageType).getRawType(), this.methodParameter);
                } catch (Exception var4) {
                    log.info("convert failed. str:{}, msgType:{}", str, this.messageType);
                    throw new RuntimeException("cannot convert message to " + this.messageType, var4);
                }
            }
        } else {
            return messageExt;
        }
    }

    private MethodParameter getMethodParameter() {
        Class targetClass;
        if (this.rocketMQListener != null) {
            targetClass = AopProxyUtils.ultimateTargetClass(this.rocketMQListener);
        } else {
            targetClass = AopProxyUtils.ultimateTargetClass(this.rocketMQReplyListener);
        }

        Type messageType = this.getMessageType();
        Class clazz = null;
        if (messageType instanceof ParameterizedType && this.messageConverter instanceof SmartMessageConverter) {
            clazz = (Class)((ParameterizedType)messageType).getRawType();
        } else {
            if (!(messageType instanceof Class)) {
                throw new RuntimeException("parameterType:" + messageType + " of onMessage method is not supported");
            }

            clazz = (Class)messageType;
        }

        try {
            Method method = targetClass.getMethod("onMessage", clazz);
            return new MethodParameter(method, 0);
        } catch (NoSuchMethodException var5) {
            var5.printStackTrace();
            throw new RuntimeException("parameterType:" + messageType + " of onMessage method is not supported");
        }
    }

    private Type getMessageType() {
        Class targetClass;
        if (this.rocketMQListener != null) {
            targetClass = AopProxyUtils.ultimateTargetClass(this.rocketMQListener);
        } else {
            targetClass = AopProxyUtils.ultimateTargetClass(this.rocketMQReplyListener);
        }

        Type matchedGenericInterface;
        Type[] actualTypeArguments;
        for(matchedGenericInterface = null; Objects.nonNull(targetClass); targetClass = targetClass.getSuperclass()) {
            actualTypeArguments = targetClass.getGenericInterfaces();
            if (Objects.nonNull(actualTypeArguments)) {
                Type[] var4 = actualTypeArguments;
                int var5 = actualTypeArguments.length;

                for(int var6 = 0; var6 < var5; ++var6) {
                    Type type = var4[var6];
                    if (type instanceof ParameterizedType && (Objects.equals(((ParameterizedType)type).getRawType(), RocketMQListener.class) || Objects.equals(((ParameterizedType)type).getRawType(), RocketMQReplyListener.class))) {
                        matchedGenericInterface = type;
                        break;
                    }
                }
            }
        }

        if (Objects.isNull(matchedGenericInterface)) {
            return Object.class;
        } else {
            actualTypeArguments = ((ParameterizedType)matchedGenericInterface).getActualTypeArguments();
            if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) {
                return actualTypeArguments[0];
            } else {
                return Object.class;
            }
        }
    }

    private void initRocketMQPushConsumer() throws MQClientException {
        if (this.rocketMQListener == null && this.rocketMQReplyListener == null) {
            throw new IllegalArgumentException("Property 'rocketMQListener' or 'rocketMQReplyListener' is required");
        } else {
            Assert.notNull(this.consumerGroup, "Property 'consumerGroup' is required");
            Assert.notNull(this.nameServer, "Property 'nameServer' is required");
            Assert.notNull(this.topic, "Property 'topic' is required");
            RPCHook rpcHook = RocketMQUtil.getRPCHookByAkSk(this.applicationContext.getEnvironment(), this.rocketMQMessageListener.accessKey(), this.rocketMQMessageListener.secretKey());
            boolean enableMsgTrace = this.rocketMQMessageListener.enableMsgTrace();
            if (Objects.nonNull(rpcHook)) {
                this.consumer = new DefaultMQPushConsumer(this.consumerGroup, rpcHook, new AllocateMessageQueueAveragely(), enableMsgTrace, this.applicationContext.getEnvironment().resolveRequiredPlaceholders(this.rocketMQMessageListener.customizedTraceTopic()));
                this.consumer.setVipChannelEnabled(false);
            } else {
                log.debug("Access-key or secret-key not configure in " + this + ".");
                this.consumer = new DefaultMQPushConsumer(this.consumerGroup, enableMsgTrace, this.applicationContext.getEnvironment().resolveRequiredPlaceholders(this.rocketMQMessageListener.customizedTraceTopic()));
            }

            this.consumer.setNamespace(this.namespace);
            String customizedNameServer = this.applicationContext.getEnvironment().resolveRequiredPlaceholders(this.rocketMQMessageListener.nameServer());
            if (customizedNameServer != null) {
                this.consumer.setNamesrvAddr(customizedNameServer);
            } else {
                this.consumer.setNamesrvAddr(this.nameServer);
            }

            if (this.accessChannel != null) {
                this.consumer.setAccessChannel(this.accessChannel);
            }

            this.consumer.setConsumeThreadMax(this.consumeThreadMax);
            this.consumer.setConsumeThreadMin(this.consumeThreadNumber);
            this.consumer.setConsumeTimeout(this.consumeTimeout);
            this.consumer.setMaxReconsumeTimes(this.maxReconsumeTimes);
            this.consumer.setAwaitTerminationMillisWhenShutdown(this.awaitTerminationMillisWhenShutdown);
            this.consumer.setInstanceName(this.instanceName);
            switch (this.messageModel) {
                case BROADCASTING:
                    this.consumer.setMessageModel(org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel.BROADCASTING);
                    break;
                case CLUSTERING:
                    this.consumer.setMessageModel(org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel.CLUSTERING);
                    break;
                default:
                    throw new IllegalArgumentException("Property 'messageModel' was wrong.");
            }

            switch (this.selectorType) {
                case TAG:
                    this.consumer.subscribe(this.topic, this.selectorExpression);
                    break;
                case SQL92:
                    this.consumer.subscribe(this.topic, MessageSelector.bySql(this.selectorExpression));
                    break;
                default:
                    throw new IllegalArgumentException("Property 'selectorType' was wrong.");
            }

            switch (this.consumeMode) {
                case ORDERLY:
                    this.consumer.setMessageListener(new DefaultMessageListenerOrderly());
                    break;
                case CONCURRENTLY:
                    this.consumer.setMessageListener(new DefaultMessageListenerConcurrently());
                    break;
                default:
                    throw new IllegalArgumentException("Property 'consumeMode' was wrong.");
            }

            this.consumer.setUseTLS(Boolean.getBoolean(this.tlsEnable));
            if (this.rocketMQListener instanceof RocketMQPushConsumerLifecycleListener) {
                ((RocketMQPushConsumerLifecycleListener)this.rocketMQListener).prepareStart(this.consumer);
            } else if (this.rocketMQReplyListener instanceof RocketMQPushConsumerLifecycleListener) {
                ((RocketMQPushConsumerLifecycleListener)this.rocketMQReplyListener).prepareStart(this.consumer);
            }

        }
    }

    public class DefaultMessageListenerOrderly implements MessageListenerOrderly {
        public DefaultMessageListenerOrderly() {
        }

        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
            Iterator var3 = msgs.iterator();

            while(var3.hasNext()) {
                MessageExt messageExt = (MessageExt)var3.next();
                DefaultRocketMQListenerContainer.log.debug("received msg: {}", messageExt);

                try {
                    long now = System.currentTimeMillis();
                    DefaultRocketMQListenerContainer.this.handleMessage(messageExt);
                    long costTime = System.currentTimeMillis() - now;
                    DefaultRocketMQListenerContainer.log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
                } catch (Exception var9) {
                    DefaultRocketMQListenerContainer.log.warn("consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}", new Object[]{messageExt.getMsgId(), messageExt.getTopic(), messageExt.getReconsumeTimes(), var9});
                    context.setSuspendCurrentQueueTimeMillis(DefaultRocketMQListenerContainer.this.suspendCurrentQueueTimeMillis);
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            }

            return ConsumeOrderlyStatus.SUCCESS;
        }
    }

    public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently {
        public DefaultMessageListenerConcurrently() {
        }

        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            Iterator var3 = msgs.iterator();

            while(var3.hasNext()) {
                MessageExt messageExt = (MessageExt)var3.next();
                DefaultRocketMQListenerContainer.log.debug("received msg: {}", messageExt);

                try {
                    long now = System.currentTimeMillis();
                    DefaultRocketMQListenerContainer.this.handleMessage(messageExt);
                    long costTime = System.currentTimeMillis() - now;
                    DefaultRocketMQListenerContainer.log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
                } catch (Exception var9) {
                    DefaultRocketMQListenerContainer.log.warn("consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}", new Object[]{messageExt.getMsgId(), messageExt.getTopic(), messageExt.getReconsumeTimes(), var9});
                    context.setDelayLevelWhenNextConsume(DefaultRocketMQListenerContainer.this.delayLevelWhenNextConsume);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    }
}
